package cn.jly.flink.kafka;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Properties;

/**
 * write data to Kafka
 *
 * @author lanyangji
 */
public class KafkaUtils {
    /**
     * kafka集群
     */
    private static final String BROKER_LIST = "hadoop101:9092,hadoop102:9092,hadoop103:9092";
    /**
     * kafka topic
     */
    private static final String TOPIC = "first";

    /**
     * write data to kafka
     *
     * @throws Exception
     */
    public static void writeToKafka() throws Exception {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", BROKER_LIST);
        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", StringSerializer.class.getName());

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        Metric metric = new Metric();
        metric.setName("lanyangji");
        metric.setTimestamp(System.currentTimeMillis());

        HashMap<String, String> tags = new HashMap<>(2);
        HashMap<String, Object> fields = new HashMap<>(4);
        tags.put("cluster", "lanyangji");
        tags.put("host_ip", "192.168.88.101");
        fields.put("used_percent", 90d);
        fields.put("max", 27244873d);
        fields.put("used", 17244873d);
        fields.put("init", 27244873d);

        metric.setTags(tags);
        metric.setFields(fields);

        // write to kafka
        kafkaProducer.send(new ProducerRecord<>(TOPIC, JSON.toJSONString(metric)));
        System.out.println("write to kafka: " + JSON.toJSONString(metric));

        kafkaProducer.flush();
    }

    public static void main(String[] args) {
        while (true) {
            try {
                Thread.sleep(300);

                // write data to kafka
                writeToKafka();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
